package com.example.springboottest.example.kafka.simple.protostuff;

import com.example.springboottest.example.kafka.simple.ProducerinterceptorPrefix;
import com.example.springboottest.example.kafka.simple.protostuff.StudentProtostuffSerializer;
import com.example.springboottest.student.model.Student;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/*
发送消息主要有三种模式
发后即忘(fire-and-forget)--->send()    速度最快，可靠性最差
同步(sync)--->
异步(async)--->

消息在真正发往 Kafka 之前，有可能需要经历拦截器 Clnterceptor ）、
序列化器 Serializer ）和分区器（ Partitioner ）

KafkaProducer 言，它是线程安全的，我们可以在多线程的环境中复用它
 */
public class ProducerFastStartStudent {
    public static final String brokerList = "localhost:9092";
    public static final String topic = "topic-demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StudentProtostuffSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StudentProtostuffSerializer.class.getName());
        /**
         * bootstrap.servers ：该参数用来指定生产者客户端连接 Kafka 集群所需的 broker
         * 地址清单，具体的内容格式为 hostl:portl,host2:port2 ，可以设置多个地址，中间以逗号隔开
         */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        /**
         * 对于可重试的异常，如果配置了 retries 参数，那么只要在规定的重试次数内自行恢复
         * 了，就不会抛出异常 retries 参数的默认值为 ，
         */
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);

        /**
         * 自定义拦截器
         * KafkaProducer 中不仅可以指定一个拦截器，还可以指定多个拦截器以形成拦截链。拦截链
         * 会按照 interceptor.classes 参数配置的拦截器的顺序来一一执行（配置的时候，各个拦
         * 截器之间使用逗号隔开）
         */
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                ProducerinterceptorPrefix.class.getName());

        /**
         * ACKS
         * 这个参数用来指定分区中必须要有多少个副本收到这条消息
         * acks =1  默认值即为1。生产者发送消息之后，只要分区的 eader 副本成功写入消息，那么它就会收到来自服务端的成功响应
         * acks =0  生产者发送消息之后不需要等待任何服务端的响应
         * acks =-1或all  。生产者在消 息发送之后，需要等待 ISR 中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应
         */
        properties.put(ProducerConfig.ACKS_CONFIG, "1");

        /**
         * max.request.size
         * 这个参数用来限制生产者客户端能发送的消息的最大值，默认值为 1048576B ，即 1MB
         */

        /**
         * compression.type
         * 这个参数用来指定消息的压缩方式，默认值为“none ”，即默认情况下，消息不会被压缩。
         * 该参数还可以配置为gzip、snappy 和lz4
         */

        /**
         * connections.max.idle.ms
         * 用来指定在多久之后关闭限制的连接，默认值是 540000 ms ，即9分钟。
         */


        //自己直生产者客户端参数并创建 KafkaProducer 实例
        KafkaProducer<String, Student> producer = new KafkaProducer<>(properties);

        //构建所需妥发送的消息
        ProducerRecord<String, Student> record = new ProducerRecord<>(topic, new Student("001", "张三", 19, "武汉"));
        //发送消息
        try {
            //第一种方式：发后即忘(fire-and-forget)
            // producer.send(record);
            //第二种方式：同步(sync)
//            RecordMetadata metadata = producer.send(record).get();
//            System.out.println(metadata.topic() + "-" + metadata.partition() + ":" + metadata.offset());
            //第三种方式：异步(async)
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null)
                        exception.printStackTrace();
                    else
                        System.out.println(metadata.topic() + "-" + metadata.partition() + ":" + metadata.offset());
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        //关闭生产者客户端示例,提供了一个带超时时间的 close()方法，
        producer.close(1, TimeUnit.MINUTES);
    }
}